const logger = require('../../logger');
const { MessageBus } = require('../index');
const utility = require('../../utility');
class MessageSubscriber extends MessageBus {
  constructor(exchangeName) {
    super();
    this.exchangeName = exchangeName;
    this.queue = null;
    this.topics = new Set();

    this.on('connection error', async () => {
      logger.debug('MessagePublisher will be tried again in 10 seconds...');
      await utility.takeRest(10000);
      try {
        await this.initQueue();
        logger.debug('MessageSubscriber reconnect success');
        await utility.arrayChunkOperation([...this.topics.values()], 10, (key) => this.subscribe(key));
      } catch (err) {
        this.reset();
        logger.error('MessageSubscriber reconnect failed', err);
        this.emit('connection error');
      }
    });
  }

  async initQueue() {
    await this.init();
    const ret = await this.channel.assertQueue('', {
      exclusive: true, // once the consumer connection is closed, the queue should be deleted.
    });
    this.queue = ret.queue;
    await this.channel.consume(this.queue, (msg) => {
      this.emit('message', msg.fields.routingKey, msg.content.toString());
    }, {
      exclusive: true, // 单消费者
      noAck: true,
    });
  }

  async subscribe(routingKey) {
    await this.channel.bindQueue(this.queue, this.exchangeName, routingKey);
    this.topics.add(routingKey);
  }

  async unsubscribe(routingKey) {
    await this.channel.unbindQueue(this.queue, this.exchangeName, routingKey);
    this.topics.delete(routingKey);
  }

  async end() {
    await this.connection.close();
  }
}

module.exports = MessageSubscriber;
